package com.atgouwu.scmall.order.service.impl;

import com.alibaba.fastjson2.JSONArray;
import com.alicp.jetcache.anno.CacheType;
import com.alicp.jetcache.anno.Cached;
import com.atgouwu.scmall.order.entity.OrderItemEntity;
import com.rabbitmq.client.Channel;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;
import com.alibaba.fastjson2.JSONObject; //fastjson2
import java.util.ArrayList; //ArrayList
import java.util.List; //List
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.atgouwu.scmall.common.utils.PageUtils;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page; //Page(删除Query)
import com.atgouwu.scmall.order.dao.OrderDao;
import com.atgouwu.scmall.order.entity.OrderEntity;
import com.atgouwu.scmall.order.service.OrderService;

/**
 * 整合Mybatis-plus
 */
@RabbitListener(queues = {"test.release.test.queue"}) //**(实现类上或方法上)监听的队列名,同一个消息只能被一个客户端收到,只有消息被消费且方法运行后,才能接收到下一条消息
@Service("orderService")
public class OrderServiceImpl extends ServiceImpl<OrderDao, OrderEntity> implements OrderService {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    /**
     * 阿里JetCache缓存: 缓存中没有就执行该方法,有就不执行直接返回
     *                 key为name+key: name随意(最好见名知意,类名:方法名:), name="“(可为空,即只有key), 也可不要name(会默认类全名),用作key前缀
     *                                key必须: key="'最好英文字符串(可中文)'", 或"#参数名", 或不要key(会默认参数名), 不能key="",不然无法存入redis
     *                 过期时间: expire(时长)+timeUnit(单位)
     *                 缓存类型: cacheType=CacheType.REMOTE默认远程, CacheType.LOCAL本地, CacheType.BOTH组合成二级缓存
     */
    @Cached(name="RedisServiceImpl:jetCache:", key="#id", expire=30, timeUnit= TimeUnit.MINUTES, cacheType= CacheType.REMOTE)
    @Override
    public String jetCache(Long id) {
        List<OrderEntity> list = new ArrayList<>();
        OrderEntity orderEntity = new OrderEntity();
        orderEntity.setId(id);
        list.add(orderEntity);
        return JSONArray.toJSONString(list);
    }

    /**
     * RabbitMQ发送消息: 可发送多个
     */
    @Override
    public String rabbitMq(Long id) {
        OrderEntity order = new OrderEntity(); //准备数据
        order.setId(id);
        //发送消息: 交换机名称(同绑定),路由键(同绑定),任意类型消息(对象需序列化,或json配置类),消息唯一id
        rabbitTemplate.convertAndSend("test-event-exchange", "test.release.test", order, new CorrelationData(UUID.randomUUID().toString()));
        //定时任务: 发送给死信队列,过期后会发送给普通队列,普通队列监听,执行定时任务
        rabbitTemplate.convertAndSend("test-event-exchange", "test.create.test", order, new CorrelationData(UUID.randomUUID().toString()));
        OrderItemEntity orderItemEntity = new OrderItemEntity(); //准备数据
        orderItemEntity.setId(id);
        //发送消息: 交换机名称(同绑定),路由键(同绑定),任意类型消息(对象需序列化,或json配置类),消息唯一id
        rabbitTemplate.convertAndSend("test-event-exchange", "test.release.test", orderItemEntity, new CorrelationData(UUID.randomUUID().toString()));
        return "rabbitMq发送成功";
    }

    /**
     * 监听(接收消息): 根据实体类型接收
     */
    @RabbitHandler //**(方法上)处理消息
    public void recieveMessage(Message message, OrderEntity order, Channel channel) { //原生消息,用对象接收(发的对象),通道
        byte[] body = message.getBody(); //消息体
        MessageProperties messageProperties = message.getMessageProperties(); //消息头
        //System.out.println("消息:" + message + ",类型:" + message.getClass() + ",消息体:" + body);
        try {
            Long deliveryTag = messageProperties.getDeliveryTag();
            channel.basicAck(deliveryTag, false); //消费者-手动签收: 标签(通道内按顺序自增),不批量签收(默认自动签收,签收后从队列移除)
            //channel.basicNack(deliveryTag, false, false); //拒收: 标签,不批量拒收(true以前全部被拒),不重新入队(丢弃,true重新入队)
            //channel.basicReject(deliveryTag, false); //拒收: 标签,不重新入队
        } catch (Exception e) {
            //网络中断
            e.printStackTrace();
        }
    }

    /**
     * 监听(接收消息): 根据实体类型接收
     *               方法重载:接收不同类型消息(不同实体类)
     */
    @RabbitHandler //**(方法上)处理消息
    public void recieveMessages(Message message, OrderItemEntity orderItemEntity, Channel channel) { //原生消息,用对象接收(发的对象),通道
        byte[] body = message.getBody(); //消息体
        MessageProperties messageProperties = message.getMessageProperties(); //消息头
        //System.out.println("消息:" + message + ",类型:" + message.getClass() + ",消息体:" + body);
        try {
            Long deliveryTag = messageProperties.getDeliveryTag();
            //channel.basicAck(deliveryTag, false); //消费者-手动签收:标签(通道内按顺序自增),不批量签收。默认自动签收,签收后从队列移除
            channel.basicNack(deliveryTag, false, false); //拒收:标签,不批量拒收(true以前全部被拒),不重新入队(丢弃,true重新入队)
            //channel.basicReject(deliveryTag, false); //拒收:标签,不重新入队
        } catch (Exception e) {
            //网络中断
            e.printStackTrace();
        }
    }

    /**
     * Kafka消息队列: 发送消息
     */
    @Override
    public String kafka(Long id) {
        //方式1:异步发送,可靠性差
        OrderEntity order = new OrderEntity(); //准备数据
        order.setId(id);
        kafkaTemplate.send("testTopic", 0, System.currentTimeMillis(), "msgId", JSONObject.toJSONString(order)); //类型,分区(必须集群),发送时间,消息id,消息字符串
        //**方式2:异步发送,回调
        String msg = "nihaomeinv";
        kafkaTemplate.send("testTopic", msg).addCallback(success -> {
            String topic = success.getRecordMetadata().topic(); //类型
            int partition = success.getRecordMetadata().partition(); //分区
            long offset = success.getRecordMetadata().offset(); //消息在分区内的偏移量
            System.out.println("发送消息成功:" + topic + "-" + partition + "-" + offset);
        }, failure -> {
            System.out.println("发送消息失败:" + failure.getMessage());
        });
        //方式3:同步发送
        try {
            kafkaTemplate.send("testTopic", msg).get(); //throws异常
        } catch (Exception e) {
            e.printStackTrace();
        }
        return "kafka发送成功";
    }

    /**
     * Kafka消息队列: 监听(接收消息)
     */
    @KafkaListener(topics = {"testTopic"}, topicPattern = "0", errorHandler = "listenerErrorHandler")  //监听:主题,分区,错误处理
    public void handMessage(ConsumerRecord<String, String> record, Acknowledgment ack, Consumer consumer, String msgs) {
        String topic = record.topic();
        String msg = record.value();
        ack.acknowledge(); //手动签收(根据配置提交偏移量),防止重新消费
        //ack.nack(100,1000); //拒收当前消息,并睡眠10秒后接收第100条后消息
        consumer.commitSync(); //手动同步提交偏移量
        //consumer.commitAsync(); //手动异步提交偏移量,防止消息丢失
        //消费者事务
        System.out.println("消费者接受消息：topic-->" + topic + ",msg->>" + msg);
    }

    //分页模糊检索
    @Override
    public PageUtils queryPage(Map<String, Object> params) {
        QueryWrapper<OrderEntity> queryWrapper = new QueryWrapper<>();
        if(params.get("name") != null && !"".equals(params.get("name"))) {
            queryWrapper.like("name", params.get("name")); //模糊检索
        }
        IPage<OrderEntity> page = this.page( //分页
                new Page<>(Long.valueOf(String.valueOf(params.get("pageNum"))),Long.valueOf(String.valueOf(params.get("pageSize")))),
                queryWrapper
        );
        List<OrderEntity> records = page.getRecords(); //解决long丢失精度,或修改id为非递增(用雪花算法)
        List<Object> vos = new ArrayList<>();
        for (OrderEntity record : records) {
                OrderEntity vo = JSONObject.parseObject(JSONObject.toJSONString(record), OrderEntity.class);
            vos.add(vo);
        }
        //return new PageUtils(vos, (int) page.getTotal(), (int) page.getSize(), (int) page.getCurrent());
        return new PageUtils(page.getRecords(), (int) page.getTotal(), (int) page.getSize(), (int) page.getCurrent());
    }

    //queryWrapper参考
    //@Override
    public int reference(OrderEntity entity) {
        //根据条件修改
        QueryWrapper<OrderEntity> queryWrapper = new QueryWrapper<>();
        queryWrapper.eq("id", entity.getId());
        //entity.setUpdateTime(new Date());
        int counta = baseMapper.update(entity, queryWrapper);
        //根据id修改
        int countb = baseMapper.updateById(entity);
        return counta;
    }

}